package com.chb.pagerank;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MyPageRankMapper extends Mapper<Text, Text, Text, Text>{
	@Override
	protected void map(Text key, Text value, Context context) throws IOException,
			InterruptedException { 
		String page = key.toString();
		Node sourceNode = null;
		/**
		 * 由于PageRank需要进行迭代计算，达到收敛
		 * 第一次统计的原始数据为A B D 
		 * PageRank设置页面的初始PR值为1.0   原始数据应为A   1.0 	B	D
		 * 进过第一轮mapreduce，计算出新的PR值  
		 * 生成数据为：  A   new PR 	B	D
		 * 同时输出A对出链节点贡献的PR值
		 * B  0.5
		 * D  0.5
		 */
		int runCount = context.getConfiguration().getInt("runCount", 0);
		//第一轮计算
		if (runCount == 0) {
			sourceNode = Node.fromMR("1.0"+"\t"+value.toString());
		}else {
		    sourceNode = Node.fromMR(value.toString());
		}
		//输出sourceNode的PR值和出链的节点
		context.write(key, new Text(sourceNode.toString()));
		
		//输出sourceNode出链节点的对应的PR值
		if (sourceNode.containsAdjacentNodes()) {
			for (int i = 0; i < sourceNode.getAdjacentNodeNames().length; i++) {
				//sourceNode分给每个出链节点的PR值为sourceNode的PR值/出链节点数目
				double pr = sourceNode.getPageRank()/sourceNode.getAdjacentNodeNames().length;
				context.write(new Text(sourceNode.getAdjacentNodeNames()[i]), new Text(pr+""));
			}
		}
	}
}
